package com.zhuliwei.rabbitmq.direct.consumer;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.zhuliwei.constants.SystemConstants;
import com.zhuliwei.domain.entity.Article;
import com.zhuliwei.domain.entity.Notice;
import com.zhuliwei.domain.entity.User;
import com.zhuliwei.mapper.UserMapper;
import com.zhuliwei.service.NoticeService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.zhuliwei.constants.RabbitConstants.NOTICE_QUEUE3_NAME;

/**
 * Article文章消息消费
 *
 * @author 朱力维
 * @time 2024/11/14 15:15
 **/
@Component
@Slf4j
@RabbitListener(queues = NOTICE_QUEUE3_NAME)
public class NoticeArticleConsumer {
    @Resource
    private NoticeService noticeService;
    @Autowired
    private UserMapper userMapper;
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * 有新文章发布通知所有用户
     *
     * @param article
     */
    @RabbitHandler
// 使用线程池通知所有用户
    public void addArticleNotice(Article article) {
        System.out.println("CPU数：" + Runtime.getRuntime().availableProcessors());
        LambdaQueryWrapper<User> lqw = new LambdaQueryWrapper<>();
        lqw.select(User::getId, User::getLogintime);
        lqw.eq(User::getType, SystemConstants.TYPE_USER);
        //只通知活跃用户（七天内登录过）
        lqw.isNotNull(User::getLogintime);
        LocalDateTime sevenDayAgo = LocalDateTime.now().minusDays(7);
        lqw.ge(User::getLogintime, sevenDayAgo);
        List<User> users = userMapper.selectList(lqw);
        // 创建一个固定大小的线程池，根据你的需求调整线程数
        ExecutorService executorService = Executors.newFixedThreadPool(CPU_COUNT * 2);
        // 分批处理用户列表
        final int BATCH_SIZE = 1000; // 根据你的数据库配置调整批量大小
        // 获取开始时间
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < users.size(); i += BATCH_SIZE) {
            List<User> batchUsers = users.subList(i, Math.min(i + BATCH_SIZE, users.size()));
            executorService.submit(() -> {
                List<Notice> notices = new ArrayList<>();
                batchUsers.forEach(user -> {
                    Notice notice = new Notice();
                    notice.setContent(SystemConstants.NOTICE_ARTICLE_PREFIX + "《" + article.getTitle() + "》");
                    notice.setType("2");
                    notice.setArticleId(article.getId());
                    notice.setUserId(user.getId());
                    notice.setCreateBy(article.getCreateBy());
                    notice.setCreateTime(new Date());
//                    noticeMapper.insert(notice);
                    notices.add(notice);
                });
                // 批量插入
                noticeService.saveBatch(notices);
            });
        }
        ;
        // 关闭线程池，等待所有任务完成
        executorService.shutdown();
        while (!executorService.isTerminated()) {
            // 等待所有任务完成
        }
        // 获取结束时间
        long endTime = System.currentTimeMillis();
        // 计算消耗的时间（秒）
        long duration = (endTime - startTime) / 1000;
        System.out.println("NoticeArticleConsumer执行时间: " + duration + " 秒");//6.2万数据有线程池无批量插入74秒 java线程池+批量插入10秒 批量插入无线程池19秒 无批量插入无线程池329秒
        log.info("NoticeArticleConsumer执行时间: " + duration + " 秒");
    }
}
